package com.atguigu.flink.chapter07;

import com.atguigu.flink.chapter05.Source.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/1/20 14:06
 */
public class Flink12_TimeWindow_SourceCodeAnalysis {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 8888)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2]));
                    }
                });

        KeyedStream<WaterSensor, String> sensorKS = sensorDS.keyBy(sensor -> sensor.getId());

        WindowedStream<WaterSensor, String, TimeWindow> sensorWS = sensorKS.window(TumblingEventTimeWindows.of(Time.seconds(5)));

        sensorWS
                .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                        long start = context.window().getStart();
                        long end = context.window().getEnd();
                        out.collect("==============================================\n" +
                                "窗口为:[" + start + "," + end + ")\n" +
//                                "数据:" + elements + "\n" +
                                "数据条数=" + elements.spliterator().estimateSize() + "\n" +
                                "=======================================\n\n");
                    }
                })
                .print();
        env.execute();
    }

}
/*
以 事件时间的 滚动窗口为例：

    1、窗口怎么划分？开始时间、结束时间？
        1）开始时间 = timestamp - (timestamp - offset + windowSize) % windowSize
                    对 窗口长度 取整数倍（以1970年1月1日0点 为基准 => 因为是时间戳）
                    并不是以第一条数据 作为窗口的 起始点
        2） 结束时间 = 开始时间 + 窗口长度

    2、窗口为什么是左闭右开？
            属于窗口的最大时间戳： maxTimestamp = end - 1ms
	             => [0,10) => 属于本窗口的最大时间戳为， 10s -1ms = 9999ms
	                => 所以10s这条数据，不属于本窗口，所以是开区间

    3、窗口什么时候触发计算？
            watermark >= 窗口的最大时间戳

    4、窗口的生命周期： 什么时候创建的窗口？
        1）属于本窗口的第一条数据来的时候，new出来的
        2）属于本窗口的每条数据来的时候，都会new，但是集合是 singleton类型，不会重复存放 => 也就是说，只会有一个窗口对象

    5、窗口的生命周期： 什么时候关闭窗口？
        窗口关闭时间 = 窗口最大时间戳 + allowedLateness
        注意：如果 allowedLateness = 0（不设置），那么触发和关窗的时间一样








详细分析过程：
    WindowOperator
        => assignWindows()
            long start = TimeWindow.getWindowStartWithOffset(timestamp, (globalOffset + staggerOffset) % size, size);
                        => timestamp - (timestamp - offset + windowSize) % windowSize
                         => 1 - (1 -0 + 10) % 10  =>  start = 0s
                         => 7 - (7 -0 + 10) % 10  =>  start = 0s
                         => 12 - (12 - 0 + 10) % 10 => start = 10s
                         => 这个算法，相当于，对 窗口长度 取整数倍（以1970年1月1日0点 为基准 => 因为是时间戳）

                         => 并不是以第一条数据 作为窗口的 起始点

			return Collections.singletonList(new TimeWindow(start, start + size));
			            => 每来一条数据，都会 new一个窗口对象
			            => 将窗口对象，放入一个 SingletonList，是单例的，所以不会重复
			            => 窗口结束时间 end = 计算出来的 start + 窗口长度

            // Gets the largest timestamp that still belongs to this window.
            // 获取属于本窗口的 最大时间戳
			public long maxTimestamp() {
		            return end - 1;
	        }
	             => [0,10) => 属于本窗口的最大时间戳为， 10s -1ms = 9999ms => 所以10s这条数据，不属于本窗口，所以是开区间

        => 	long cleanupTime = window.maxTimestamp() + allowedLateness;
			return cleanupTime >= window.maxTimestamp() ? cleanupTime : Long.MAX_VALUE;
			=》 窗口关闭时间 = 窗口最大时间戳 + allowedLateness

	   窗口触发分析：

	        window.maxTimestamp() <= ctx.getCurrentWatermark()
            watermark >= 窗口的最大时间戳 \
                => 比如，[0,10)的窗口，当 watermark >= 10s - 1ms = 9999ms时触发
                => 此时，watermark = 事件时间 - 等待3s - 1ms => 事件时间 = 13s时触发
 */
